package com.example.flinktest.transformation;

import com.example.flinktest.source.Event;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


public class TransFunctionUDFTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> clicks = env.fromElements(
                new Event("JunSongLi", "./home", 1000L),
                new Event("Tomas", "./cart", 200L)
        );
        //自定义过滤方法
        DataStream<Event> streamSource = clicks.filter(new FlinkFilter());
        //map函数使用lambda表达式，返回简单类型，不需要类型声明
        DataStream<String> stream = clicks.map(event -> event.url);
        //flatMap,必须通过returns明确声明返回类型
        DataStream<String> stream1 = clicks.flatMap((Event event, Collector<String> out) -> {
            out.collect(event.url);
        }).returns(Types.STRING);

        //使用显示Returns
        DataStream<Tuple2<String, Long>> stream2 = clicks.map(e -> Tuple2.of(e.user, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));
        //使用类来替代Lambda表达式
        clicks.map(new MyTuple2Mapper()).print();
        //使用域名类替代Lambda表达式
        clicks.map(new MapFunction<Event, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Event event) throws Exception {
                return Tuple2.of(event.user, 1L);
            }
        }).print();

        streamSource.print();
        stream.print();
        stream1.print();
        stream2.print();
        env.execute();
    }

    public static class FlinkFilter implements FilterFunction<Event> {

        @Override
        public boolean filter(Event event) throws Exception {
            return event.url.contains("home");
        }
    }

    private static class MyTuple2Mapper implements MapFunction<Event, Tuple2<String, Long>> {
        @Override
        public Tuple2<String, Long> map(Event event) throws Exception {
            return Tuple2.of(event.user, 1L);
        }
    }
}
